package com.laycoding.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * created with IntelliJ IDEA.
 *
 * @author: guidada
 * @date: 2021/04/18/16:46
 * @description:
 */
@Slf4j
@Component
public class ConsumerListener {
    /** 
     * 
     * @author guidada
     * @date 2021/4/18 0018 
     * @param record 
     * @return void
     * @description ~
     */
    @KafkaListener(topics = "testTopic")
    public void consumeMessage(ConsumerRecord<?,?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()){
            Object message = kafkaMessage.get();
            log.info("----------------- record =" + record);
            log.info("----------------- message =" + message);
        }
    }
}
